package com.roy.HadoopRPCDemo.protocol.impl;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;

import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.roy.HadoopRPCDemo.bean.RPCResBean;
import com.roy.HadoopRPCDemo.bean.UserBean;
import com.roy.HadoopRPCDemo.protocol.MyService;
import com.roy.HadoopRPCDemo.util.ExecutorManager;
import com.roy.HadoopRPCDemo.util.TextArrayWritable;
import com.roy.HadoopRPCDemo.util.UserBeanArrayWritable;

import scala.Tuple2;

public class MyServiceImpl implements MyService {

	private Logger logger = LoggerFactory.getLogger(this.getClass());

	public String baseTypeService(int userId, String userName, String userPassword) {
		String res = "received UserInfo: userId => " + userId + ", userName => " + userName + ";userPassword => "
				+ userPassword;
		logger.info(res);
		return res;
	}

	public UserBean mapTypeService(MapWritable userinfo) {
		logger.info("get RPC call mapTypeService");
		String userId = ((Text) userinfo.get(new Text("userId"))).toString();
		String userName = ((Text) userinfo.get(new Text("userName"))).toString();
		String userPassword = ((Text) userinfo.get(new Text("userPassword"))).toString();

		UserBean us = new UserBean();
		us.setUserId(Integer.parseInt(userId));
		us.setUserName(userName);
		us.setUserPassword(userPassword);
		us.setUserInfo("set from server");

		return us;
	}

	public UserBean objTypeService(UserBean user) {
		logger.info("get RPC call objTypeService");
		user.setUserInfo("set from server");
		return user;
	}

	public UserBeanArrayWritable objArrayService(UserBeanArrayWritable users) {
		logger.info("get RPC call objArrayService");
		Writable[] allUser = users.get();
		for (Writable data : allUser) {
			UserBean user = (UserBean) data;
			user.setUserInfo("set from server");
		}
		return users;
	}

	public RPCResBean frameworkService(String exeId, TextArrayWritable params) {
		logger.info("get RPC call frameworkService exeId=> " + exeId);
		RPCResBean res = null;
		Tuple2<Object, Method> methodAndObj = ExecutorManager.getMethod(exeId);
		String[] strParams = convertParam(params);
		if (null != methodAndObj) {
			Object obj = methodAndObj._1;
			Method method = methodAndObj._2;

			try {
				Object invoke = method.invoke(obj, strParams);
				res = (RPCResBean)invoke;
			} catch (Exception e) {
				logger.error("execute service failed exeId=> " + exeId+";e: "+e.getMessage());
				e.printStackTrace();
//			} catch (IllegalArgumentException e) {
//				logger.error("execute service failed exeId=> " + exeId+";e: "+e.getMessage());
//			} catch (InvocationTargetException e) {
//				logger.error("execute service failed exeId=> " + exeId+";e: "+e.getMessage());
			} 

		}
		return null==res?null:res;
	}
	
	private String[] convertParam(TextArrayWritable params) {
		Writable[] writableParams = params.get();
		String[] strParams = new String[writableParams.length];
		int index = 0;
		for (Writable data : writableParams) {
			Text txData = (Text) data;
			strParams[index++] = txData.toString();
		}
		return strParams;
	}

}
